package com.bw.gmall.realtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: DwsTrafficVcChArIsNewPageViewWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/28 18:57
 * @version: 1.0
 * 流量域版本-渠道-地区-访客类别粒度页面浏览各窗口汇总表
 */

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW, 4, 10022);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<JSONObject> etlStream = getJsonObjectSingleOutputStreamOperator(dataStreamSource);
        SingleOutputStreamOperator<TrafficPageViewBean> keyByStream = keyByStream(etlStream);
//        keyByStream.print();
        SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream = getWatermarksStream(keyByStream);
//        watermarksStream.print();
        SingleOutputStreamOperator<TrafficPageViewBean> process = getProcess(watermarksStream);
//        process.print();
        process.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW));
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> getProcess(SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream) {
        return watermarksStream.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4 getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return Tuple4.of(trafficPageViewBean.getVc(), trafficPageViewBean.getCh(), trafficPageViewBean.getAr(), trafficPageViewBean.getIsNew());
            }
        }).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
                .process(new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void process(Tuple4<String, String, String, String> tuple4, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>.Context context, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                ArrayList<TrafficPageViewBean> toList = ListUtil.toList(iterable);
                Long uvCt = toList.stream().mapToLong(e -> e.getUvCt()).sum();
                Long svCt = toList.stream().mapToLong(e -> e.getSvCt()).sum();
                Long pvCt = toList.stream().mapToLong(e -> e.getPvCt()).sum();
                long DurSum = toList.stream().mapToLong(e -> e.getDurSum()).sum();
                long start = context.window().getStart();
                String start_time = DateFormatUtil.tsToDateTime(start);
                long end = context.window().getEnd();
                String end_time = DateFormatUtil.tsToDateTime(end);
                String cur_date = (DateFormatUtil.tsToDate(System.currentTimeMillis()));
                TrafficPageViewBean firstBean = toList.iterator().next();

                TrafficPageViewBean result = TrafficPageViewBean.builder()
                        .stt(start_time)
                        .edt(end_time)
                        .cur_date(cur_date)
                        .vc(firstBean.getVc())
                        .ch(firstBean.getCh())
                        .ar(firstBean.getAr())
                        .isNew(firstBean.getIsNew())
                        .uvCt(uvCt)
                        .svCt(svCt)
                        .pvCt(pvCt)
                        .durSum(DurSum)
                        .ts(firstBean.getTs())
                        .sid(firstBean.getSid())
                        .build();
                collector.collect(result);

            }
        });
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> getWatermarksStream(SingleOutputStreamOperator<TrafficPageViewBean> keyByStream) {
        return keyByStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(10)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                return trafficPageViewBean.getTs();
            }
        }));
    }


    private static SingleOutputStreamOperator<TrafficPageViewBean> keyByStream(SingleOutputStreamOperator<JSONObject> etlStream) {
        return etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                String mid = jsonObject.getJSONObject("common").getString("mid");
                return mid;
            }
        }).process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("last_dt", String.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(24)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                //取出ts
                Long ts = jsonObject.getLong("ts");
                //转换时间格式
                String s = DateFormatUtil.tsToDate(ts);
                //取出状态的值
                String value = valueState.value();
                //uv
                Long uvCt = 0L;
                //sv
                Long svCt = 0L;
                //计算uv
                if (!s.equals(value)) {
                    uvCt = 1L;
                    valueState.update(s);
                }
                //计算sv
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_Id = page.getString("last_page_id");
                if (last_page_Id == null) {
                    svCt = 1L;
                }
                JSONObject common = jsonObject.getJSONObject("common");
                collector.collect(TrafficPageViewBean.builder()
                        .ar(common.getString("ar"))
                        .ch(common.getString("ch"))
                        .vc(common.getString("vc"))
                        .isNew(common.getString("is_new"))
                        .uvCt(uvCt)
                        .svCt(svCt)
                        .pvCt(1L)
                        .durSum(page.getLong("during_time"))
                        .ts(ts)
                        .sid(common.getString("sid"))
                        .build());
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getJsonObjectSingleOutputStreamOperator(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                JSONObject page = jsonObject.getJSONObject("page");
                if (common != null && page != null) {
                    String mid = common.getString("mid");
                    if (mid != null && ts != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });
    }
}
